
                /*********************************************
                **                                          **
                **                                          **
                **               sa.poll.cmr.C              **
                **                                          **
                **                                          **
                *********************************************/
                /*********************************************
                **                                          **
                **         Author Jean-Marc Adamo           **
                **       Copyright (C) 1997-2008 JMA        **
                *********************************************/


/*
** Spread Arrays and 
** Non-Blocking Global Read/Write Functions
** Inter Group Communication
** Needs 4 processors
*/

#include "arch.h"

Global_id gid1;
Global_id gid2;
Global_id gid3;
Coll_id cid0, cid1, cid2, cid3, cid4, cid5, cid6;

void init_gid(int, char **){
   gid1.set();
   gid2.set();
   gid3.set();
   cid0.set();
   cid1.set();
   cid2.set();
   cid3.set();
   cid4.set();
   cid5.set();
   cid6.set();
   fprintf(Sdf::trace, "from node %d: gid1.tag=%d, gid2.tag=%d, gid3.tag=%d, cid0.tag=%d, cid1.tag=%d, cid2.tag=%d, cid3.tag=%d, cid4.tag=%d, cid5.tag=%d, cid6.tag=%d\n",
               Sdf::PROC_ID, gid1.get(), gid2.get(), gid3.get(), 
               cid0.get(), cid1.get(), cid2.get(), 
               cid3.get(), cid4.get(), cid5.get(), cid6.get());
   fflush(Sdf::trace);
}


#define d0 4
#define d1 5
#define d2 2

class RW_count{
   public:
      int cnt;

      RW_count(){
         cnt = 0;
      }
      void reset(){
         cnt = 0;
      }
      void more_RW(int i){
         cnt += i;
      }
};

#define wait_while(condition) while(condition) Sdf::poll();

#define RW_not_complete() (count->cnt < d0*d1*d2)

RW_count *count;

void upon_end_of_RW(HRW_rcd *){
   count->cnt++;
}

class Root_thread: public Thread{
   public:
        Root_thread()
                : Thread((Tmf0_ptr)&Root_thread::body){}

        void body(){

           fprintf(Sdf::trace, "node %d, in Root_proc::start\n", Sdf::PROC_ID);
           fflush(Sdf::trace); 

           Group *group1;
           int rk1[1];
           int i;
           for(i = 0; i < 1; i++)
              rk1[i] = i;
           group1 = Sdf::incl(Sdf::COMML_WORLD, 1, rk1);
           CommL *comm1 = group1->commit(cid1, Sdf::COMML_WORLD);

           Group *group2;
           int *rk2 = new int[Sdf::PROC_NB-1];
           for(i = 1; i < Sdf::PROC_NB; i++)
              rk2[i-1] = i;
           group2 = Sdf::incl(Sdf::COMML_WORLD, Sdf::PROC_NB - 1, rk2);
           CommL *comm2 = group2->commit(cid2, Sdf::COMML_WORLD);

           Sdf::barrier(cid4);

           CommR *commr;
           commr = Sdf::newCommR(cid3, comm1, comm2, Sdf::COMML_WORLD);

           fprintf(Sdf::trace,"node%d, ######1, commr=%p, comm1=%p, comm2=%p\n",
                   Sdf::PROC_ID, 
                   commr, comm1, comm2);
           fflush(Sdf::trace);

           int *local_ranks = commr->get_local_CommL()->get_world_ranks();
           int *remote_ranks = commr->get_remote_CommL()->get_world_ranks();

           fprintf(Sdf::trace,"node%d, @@@@@@check local in commr, local_CommL=%d, size=%d, ranks=%d, %d, %d\n",
                   Sdf::PROC_ID, 
                   commr->get_local_CommL(),
                   commr->get_local_CommL()->size(),
                   local_ranks[0], local_ranks[1], local_ranks[2]);
           fflush(Sdf::trace);
           fprintf(Sdf::trace,"node%d, @@@@@@check remote in commr, remote_CommL=%d, size=%d, ranks=%d, %d, %d\n",
                   Sdf::PROC_ID, 
                   commr->get_remote_CommL(),
                   commr->get_remote_CommL()->size(),
                   remote_ranks[0], remote_ranks[1], remote_ranks[2]);
           fflush(Sdf::trace);

           delete []local_ranks;
           delete []remote_ranks;

             //access space: comm1, 
             //definition space: comm2
           SpreadArray<int> SA0(gid1, d0, d1, d2, 2, 
                                commr, comm2);
           SpreadArray<int> SA1(gid2, d0, d1, d2, 2, 
                                commr, comm2);
           SpreadArray<int> SA2(gid3, d0, d1, d2, 2, 
                                commr, comm2);

if(!Sdf::PROC_ID){
fprintf(Sdf::trace, "node%d:>>>SA0 p_id(wrt comm)=%d, offset=%d, comm=%d\n", 
        Sdf::PROC_ID, 
        SA0.get_proc_id(),
        SA0.get_offset(),
        SA0.get_comm());
fflush(Sdf::trace);

fprintf(Sdf::trace, "node%d:>>>SA1 p_id(wrt comm)=%d, offset=%d, comm=%d\n", 
        Sdf::PROC_ID, 
        SA1.get_proc_id(),
        SA1.get_offset(),
        SA1.get_comm());
fflush(Sdf::trace);

fprintf(Sdf::trace, "node%d:>>>SA2 p_id(wrt comm)=%d, offset=%d, comm=%d\n", 
        Sdf::PROC_ID, 
        SA2.get_proc_id(),
        SA2.get_offset(),
        SA2.get_comm());
fflush(Sdf::trace);
}


           if(Sdf::PROC_ID==0){

              int L0[d0][d1][d2];
              int L1[d0][d1][d2];
              int L2[d0][d1][d2];

              RW_count c;
              count = &c;

              RWCtrl ctrl(upon_end_of_RW, 0);
              SA0.set_ctrl_ptr(&ctrl);
              SA1.set_ctrl_ptr(&ctrl);
              SA2.set_ctrl_ptr(&ctrl);

              Sdf::barrier(cid5);
                //to synchronize the process in comm1
                //with the processes in comm2 before printing
                //writing-reading sequence results.

              fprintf(Sdf::trace, "node%d: !!!!!!!!!!1\n", Sdf::PROC_ID);
              int i0;
              for(i0=0; i0<d0; i0++)
                 for(int j = 0; j < d1; j++)
                    for(int k = 0; k < d2; k++)
                       SA0(i0,j,k).read(L0[i0][j][k]);
              wait_while(RW_not_complete());
              for(i0=0; i0<d0; i0++)
                 for(int j = 0; j < d1; j++)
                    for(int k = 0; k < d2; k++){
                       fprintf(Sdf::trace, "node%d: i0=%d, j=%d, k=%d, data=%d\n", 
                               Sdf::PROC_ID, i0, j, k, L0[i0][j][k]);
                       fflush(Sdf::trace);
                    }

              c.reset();
              int i1;
              for(i1=0; i1<d0; i1++)
                 for(int j = 0; j < d1; j++)
                    for(int k = 0; k < d2; k++)
                       SA1(i1,j,k).read(L1[i1][j][k]);
              wait_while(RW_not_complete());
              for(i1=0; i1<d0; i1++)
                 for(int j = 0; j < d1; j++)
                    for(int k = 0; k < d2; k++){
                       fprintf(Sdf::trace, "node%d: i1=%d, j=%d, k=%d, data=%d\n", 
                               Sdf::PROC_ID, i1, j, k, L1[i1][j][k]);
                       fflush(Sdf::trace);
                    }

              c.reset();
              int i2;
              for(i2=0; i2<d0; i2++)
                 for(int j = 0; j < d1; j++)
                    for(int k = 0; k < d2; k++)
                       SA2(i2,j,k).read(L2[i2][j][k]);
              wait_while(RW_not_complete());
              for(i2=0; i2<d0; i2++)
                 for(int j = 0; j < d1; j++)
                    for(int k = 0; k < d2; k++){
                       fprintf(Sdf::trace, "node%d: i2=%d, j=%d, k=%d, data=%d\n", 
                               Sdf::PROC_ID, i2, j, k, L2[i2][j][k]);
                       fflush(Sdf::trace);
                    }


              Sdf::barrier(cid6);


              return;
           }

              //processes in comm2 are the only ones executing the sequence below
           int *local_ptr0 = SA0.to_local_ptr();
           int *local_ptr1 = SA1.to_local_ptr();
           int *local_ptr2 = SA2.to_local_ptr();

           int d = (d0*d1*d2)/comm2->size() + 1;

           for(i = 0; i < d; i++){
              local_ptr0[i] = 10*Sdf::PROC_ID + i;
              local_ptr1[i] = 20*Sdf::PROC_ID + i;
              local_ptr2[i] = 30*Sdf::PROC_ID + i;
           }

           Sdf::barrier(cid5);
           fprintf(Sdf::trace, "node%d: !!!!!!!!!!2\n", Sdf::PROC_ID);

           Sdf::barrier(cid6);

           delete comm1;
           delete comm2;
           delete commr;

           delete []rk2;

           fprintf(Sdf::trace, "node%d: the end\n\n\n", Sdf::PROC_ID);
           fflush(Sdf::trace);
        }
};

MAIN(Root_thread, init_gid, Sdf::empty, Sdf::empty)
